package com.health.trans.processor.reader;

import com.health.trans.annotation.Reader;
import com.health.trans.common.util.GenericUtil;
import com.health.trans.kit.KafkaReaderConfig;
import com.health.trans.kit.ReaderConfig;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.kafka.KafkaItemReader;
import org.springframework.batch.item.kafka.builder.KafkaItemReaderBuilder;

/**
 * @author nqlz
 */
@Reader
public class KafkaReader implements CsReader {


    @Override
    public <T> ItemReader<T> getReader(ReaderConfig<T> readerConfig) {
        KafkaReaderConfig<T> config = (KafkaReaderConfig<T>) readerConfig;

        KafkaItemReader<Object, Object> build = new KafkaItemReaderBuilder<>()
                .topic(config.getTopic())
                .consumerProperties(config.getProperties())
                .partitions(config.getPartitions())
                .name(config.getTableClass().getSimpleName()).build();

        return GenericUtil.get(build);
    }



}
